/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.zookeeper.server.quorum;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.jute.BinaryInputArchive;
import org.apache.jute.BinaryOutputArchive;
import org.apache.jute.InputArchive;
import org.apache.jute.OutputArchive;
import org.apache.jute.Record;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.zookeeper.server.Request;
import org.apache.zookeeper.server.ServerCnxn;
import org.apache.zookeeper.server.ZooTrace;
import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
import org.apache.zookeeper.server.util.SerializeUtils;
import org.apache.zookeeper.server.util.ZxidUtils;
import org.apache.zookeeper.txn.TxnHeader;

/**
 * This class is the superclass of two of the three main actors in a ZK
 * ensemble: Followers and Observers. Both Followers and Observers share 
 * a good deal of code which is moved into Peer to avoid duplication. 
 */
public class Learner {

    static class PacketInFlight {
        /** 事务头 */
        TxnHeader hdr;
        /** 记录 */
        Record rec;
    }

    /** 当前集群对象 */
    QuorumPeer self;
    /** 当前learner状态的zk服务器 */
    LearnerZooKeeperServer zk;
    
    protected BufferedOutputStream bufferedOutput;
    
    protected Socket sock;
    
    /**
     * Socket getter
     * @return 
     */
    public Socket getSocket() {
        return sock;
    }

    /** 输入 */
    protected InputArchive leaderIs;
    /** 输出 */
    protected OutputArchive leaderOs;  
    /** 当前协议版本 */
    protected int leaderProtocolVersion = 0x01;
    
    protected static final Logger LOG = LoggerFactory.getLogger(Learner.class);

    /** 连接leader是否允许延迟 */
    static final private boolean nodelay = System.getProperty("follower.nodelay", "true").equals("true");

    static {
        LOG.info("TCP NoDelay set to: " + nodelay);
    }

    /**
     * client连接到learner时，learner要向leader提出REVALIDATE请求，在收到回复之前，记录在一个map中，表示尚未处理完的验证
     */
    final ConcurrentHashMap<Long, ServerCnxn> pendingRevalidations = new ConcurrentHashMap<>();
    
    public int getPendingRevalidationsCount() {
        return pendingRevalidations.size();
    }
    
    /**
     * 集群版client重连时调用，learner验证会话是否有效，并激活，需要发送请求给Leader
     * @param clientId the client to be revalidated
     * @param timeout the timeout for which the session is valid
     * @return
     * @throws IOException
     */
    void validateSession(ServerCnxn cnxn, long clientId, int timeout) throws IOException {
        LOG.info("Revalidating client: 0x" + Long.toHexString(clientId));
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        DataOutputStream dos = new DataOutputStream(baos);
        dos.writeLong(clientId); // 写入clientId
        dos.writeInt(timeout);
        dos.close();
        // 发送REVALIDATE命令
        QuorumPacket qp = new QuorumPacket(Leader.REVALIDATE, -1, baos.toByteArray(), null);
        // 需要验证，还未返回结果，记录在map中
        pendingRevalidations.put(clientId, cnxn);
        // 发送
        writePacket(qp, true);
    }     
    
    /**
     *  发送packet，record的tag是packet，让leader读
     * @param pp the proposal packet to be sent to the leader
     * @throws IOException
     */
    void writePacket(QuorumPacket pp, boolean flush) throws IOException {
        synchronized (leaderOs) {
            if (pp != null) {
                // 调用 QuorumPacket 的 serialize()方法
                leaderOs.writeRecord(pp, "packet"); // 基于Leader的jute输出流
                // 将QuorumPacket这个对象进行序列化之后，以字节流发送到Leader
            }
            if (flush) {
                bufferedOutput.flush();
            }
        }
    }

    /**
     * 从leader读取packet
     * @param pp the packet to be instantiated
     * @throws IOException
     */
    void readPacket(QuorumPacket pp) throws IOException {
        synchronized (leaderIs) {
            // 如果读到数据，会放到 QuorumPacket里面
            leaderIs.readRecord(pp, "packet");
        }

        long traceMask = ZooTrace.SERVER_PACKET_TRACE_MASK;
        if (pp.getType() == Leader.PING) {
            traceMask = ZooTrace.SERVER_PING_TRACE_MASK;
        }
    }
    
    /**
     * 所有请求都被LEADER处理，learner接收到请求会转发给Leader
     * @param request the request from the client
     * @throws IOException
     */
    void request(Request request) throws IOException {
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        DataOutputStream oa = new DataOutputStream(baos);
        oa.writeLong(request.sessionId);
        oa.writeInt(request.cxid);
        oa.writeInt(request.type);
        if (request.request != null) {
            request.request.rewind();
            int len = request.request.remaining();
            byte b[] = new byte[len];
            request.request.get(b);
            request.request.rewind();
            oa.write(b);
        }
        oa.close();
        QuorumPacket qp = new QuorumPacket(Leader.REQUEST, -1, baos
                .toByteArray(), request.authInfo);
        writePacket(qp, true);
    }
    
    /**
     * 找到leader是谁，就是通过currentVote的sid遍历所有集群机器，看哪个sid一样，就是那台机器
     */
    protected InetSocketAddress findLeader() {
        InetSocketAddress addr = null;
        Vote current = self.getCurrentVote();
        for (QuorumServer s : self.getView().values()) {
            // 集群中某个机器的sid和当前投票的sid一样
            if (s.id == current.getId()) {
                addr = s.addr;
                break;
            }
        }
        if (addr == null) {
            LOG.warn("Couldn't find the leader with id = " + current.getId());
        }
        return addr;
    }
    
    /**
     * 和leader建立连接. 尝试5次.
     * @param addr - the address of the Leader to connect to.
     * @throws IOException - if the socket connection fails on the 5th attempt
     * @throws ConnectException
     * @throws InterruptedException
     */
    protected void connectToLeader(InetSocketAddress addr)  throws IOException, ConnectException, InterruptedException {
        sock = new Socket();        
        sock.setSoTimeout(self.tickTime * self.initLimit);
        // 重试5次
        for (int tries = 0; tries < 5; tries++) {
            try {
                // 走TCP三次握手，完成连接的建立
                sock.connect(addr, self.tickTime * self.syncLimit);
                // 如果连接建立之后，双方就有一个socket通信
                sock.setTcpNoDelay(nodelay);
                break;
            } catch (IOException e) {
                if (tries == 4) {
                    LOG.error("Unexpected exception",e);
                    throw e;
                } else {
                    LOG.warn("Unexpected exception, tries="+tries+ ", connecting to " + addr,e);
                    sock = new Socket();
                    sock.setSoTimeout(self.tickTime * self.initLimit);
                }
            }
            Thread.sleep(1000);
        }
        // zk的序列化和反序列化 用的是 jute
        // 获取对leader的输入流，缓冲输入流，archive输入流
        leaderIs = BinaryInputArchive.getArchive(new BufferedInputStream( sock.getInputStream()));
        // 缓冲输出流, archive缓冲输出流
        bufferedOutput = new BufferedOutputStream(sock.getOutputStream());
        leaderOs = BinaryOutputArchive.getArchive(bufferedOutput);
    }   
    
    /**
     * 连上leader后，进行握手,
     * 注册时会发把自己基本信息发送给leader，称为learnerInfo
     * @param pktType 代表一个following或者observing连接的注册
     * @return the zxid the Leader sends for synchronization purposes.
     * @throws IOException
     */
    protected long registerWithLeader(int pktType) throws IOException{
    	long lastLoggedZxid = self.getLastLoggedZxid();
        QuorumPacket qp = new QuorumPacket();
        // 设置类型为Leader.FOLLOWERINFO或者Leader.OBSERVERINFO
        qp.setType(pktType); // 类型
        qp.setZxid(ZxidUtils.makeZxid(self.getAcceptedEpoch(), 0)); // 最大的zxid
        
        LearnerInfo li = new LearnerInfo(self.getId(), 0x10000);
        ByteArrayOutputStream bsid = new ByteArrayOutputStream();  // 字节数据的输出流
        BinaryOutputArchive boa = BinaryOutputArchive.getArchive(bsid); // 使用jute封装输出流
        // 把learner当前信息发送给leader
        boa.writeRecord(li, "LearnerInfo");  // 先将 LeanerInfo 对象进行序列化，写入底层
        // ByteArrayOutputStream
        // 会把字节数组输出流里的LeanerInfo对象的序列化后的数据转换为字节数组，设置到QuorumPacket对象里去
        qp.setData(bsid.toByteArray());

        // 发送LearnerInfo包
        writePacket(qp, true);
        // 接收leader的回复(新版本是一个LEADERINFO的消息，包含leader的状态)
        readPacket(qp);        
        final long newEpoch = ZxidUtils.getEpochFromZxid(qp.getZxid());
        // 新版本的leader
		if (qp.getType() == Leader.LEADERINFO) {
        	leaderProtocolVersion = ByteBuffer.wrap(qp.getData()).getInt();
        	byte epochBytes[] = new byte[4];
        	final ByteBuffer wrappedEpochBytes = ByteBuffer.wrap(epochBytes);
        	if (newEpoch > self.getAcceptedEpoch()) {
        		wrappedEpochBytes.putInt((int)self.getCurrentEpoch());
        		self.setAcceptedEpoch(newEpoch);
        	} else if (newEpoch == self.getAcceptedEpoch()) {
                wrappedEpochBytes.putInt(-1);
        	} else {
        		throw new IOException("Leaders epoch, " + newEpoch + " is less than accepted epoch, " + self.getAcceptedEpoch());
        	}
            // 8.接受完了leader状态之后，要发送ACK消息
        	QuorumPacket ackNewEpoch = new QuorumPacket(Leader.ACKEPOCH, lastLoggedZxid, epochBytes, null);
        	writePacket(ackNewEpoch, true);
            return ZxidUtils.makeZxid(newEpoch, 0);
        } else { //老版本的leader(用于兼容)
        	if (newEpoch > self.getAcceptedEpoch()) {
        		self.setAcceptedEpoch(newEpoch);
        	}
            if (qp.getType() != Leader.NEWLEADER) {
                LOG.error("First packet should have been NEWLEADER");
                throw new IOException("First packet should have been NEWLEADER");
            }
            return qp.getZxid();
        }
    } 
    
    /**
     * 启动时learner先和leader进行数据同步
     *
     * 1.DIFF 和 SNAP 两种数据同步的方式的区别是什么
     *      打个比方,LEADER的样子是A，LEARNER的样式是B
     *      DIFF：代表LEADER后续会告诉LEARNER：不断接收PROPOSAL和COMMIT，一步步提交让LEARNER从B变成B1，到B2。。。最后到A的样子.
     *      SNAP：代表LEADER直接告诉LEARNER:我长得样子是A，你copy一下变成我的样子就好了
     *
     * 2.是否发送了DIFF ,SNAP或者TRUNC，learner接受到一下就一下子数据同步完成了
     *      不是
     *      DIFF：后面会跟着一堆INFORM和COMMIT让learner一步步的同步
     *      TRUNC：回滚到某个zxid后，后面也会跟着一堆INFORM和COMMIT
     *      SNAP：后面会发送自己db的序列化内容
     *
     * @param newLeaderZxid
     * @throws IOException
     * @throws InterruptedException
     */
    protected void syncWithLeader(long newLeaderZxid) throws IOException, InterruptedException{
        QuorumPacket ack = new QuorumPacket(Leader.ACK, 0, null, null);
        QuorumPacket qp = new QuorumPacket();
        long newEpoch = ZxidUtils.getEpochFromZxid(newLeaderZxid);
        
        readPacket(qp);   
        LinkedList<Long> packetsCommitted = new LinkedList<>();
        // 收到proposal但是还未commit的包
        LinkedList<PacketInFlight> packetsNotCommitted = new LinkedList<>();
        synchronized (zk) {
            // 接收diff，表示以diff方式与leader的数据同步
            if (qp.getType() == Leader.DIFF) {
                LOG.info("Getting a diff from the leader 0x" + Long.toHexString(qp.getZxid()));                
            }
            // 表示snap方式与leader同步,从leader复制一份镜像数据到本地内存
            else if (qp.getType() == Leader.SNAP) {
                LOG.info("Getting a snapshot from leader");
                zk.getZKDatabase().clear();
                //从leader复制一份镜像数据到本地内存
                zk.getZKDatabase().deserializeSnapshot(leaderIs);
                String signature = leaderIs.readString("signature");
                // 验证签名
                if (!signature.equals("BenWasHere")) {
                    LOG.error("Missing signature. Got " + signature);
                    throw new IOException("Missing signature");                   
                }
            }
            // 触发回滚，回滚到leader的lastzxid
            else if (qp.getType() == Leader.TRUNC) {
                LOG.warn("Truncating log to get in sync with the leader 0x" + Long.toHexString(qp.getZxid()));
                boolean truncated=zk.getZKDatabase().truncateLog(qp.getZxid());
            }
            else {
                LOG.error("Got unexpected packet from leader " + qp.getType() + " exiting ... " );
                System.exit(13);
            }
            zk.getZKDatabase().setlastProcessedZxid(qp.getZxid());
            zk.createSessionTracker();            
            
            long lastQueued = 0;

            // 代表是否接受到了NEWLEADER请求，此时代表接受完了leader的同步数据，拍摄快照
            boolean snapshotTaken = false;
            outerLoop:
            // 启动时数据同步,不断读取leader的数据，直到收到UPTODATE表示同步完成
            while (self.isRunning()) {
                readPacket(qp);
                switch(qp.getType()) {
                case Leader.PROPOSAL: // 接收到提议
                    PacketInFlight pif = new PacketInFlight();
                    pif.hdr = new TxnHeader();
                    pif.rec = SerializeUtils.deserializeTxn(qp.getData(), pif.hdr);
                    if (pif.hdr.getZxid() != lastQueued + 1) {
                    LOG.warn("Got zxid 0x" + Long.toHexString(pif.hdr.getZxid()) + " expected 0x" + Long.toHexString(lastQueued + 1));
                    }
                    lastQueued = pif.hdr.getZxid();
                    // 因为还没有commit，记录在notCommit的队列里面
                    packetsNotCommitted.add(pif);
                    break;
                case Leader.COMMIT: //接收到COMMIT
                    if (!snapshotTaken) { //没拍照的话
                        pif = packetsNotCommitted.peekFirst();
                        if (pif.hdr.getZxid() != qp.getZxid()) {
                            LOG.warn("Committing " + qp.getZxid() + ", but next proposal is " + pif.hdr.getZxid());
                        } else {
                            zk.processTxn(pif.hdr, pif.rec); //直接处理事务
                            packetsNotCommitted.remove(); //从未COMMIT记录中删除对应记录
                        }
                    } else {
                        packetsCommitted.add(qp.getZxid());
                    }
                    break;
                case Leader.INFORM: //observer才会拿到INFORM消息，来同步
                    TxnHeader hdr = new TxnHeader();
                    Record txn = SerializeUtils.deserializeTxn(qp.getData(), hdr);
                    zk.processTxn(hdr, txn);
                    break;
                case Leader.UPTODATE: //过半机器完成了leader验证，自己也完成了数据同步,可以跳出循环
                    if (!snapshotTaken) {
                        zk.takeSnapshot();
                        self.setCurrentEpoch(newEpoch);
                    }
                    self.cnxnFactory.setZooKeeperServer(zk);                
                    break outerLoop;
                case Leader.NEWLEADER:
                    zk.takeSnapshot(); //设置快照和当前epoch
                    self.setCurrentEpoch(newEpoch);
                    snapshotTaken = true; //已经快照了
                    writePacket(new QuorumPacket(Leader.ACK, newLeaderZxid, null, null), true);
                    break;
                }
            }
        }
        ack.setZxid(ZxidUtils.makeZxid(newEpoch, 0));
        writePacket(ack, true); //最后再发一个ack
        sock.setSoTimeout(self.tickTime * self.syncLimit);
        zk.startup(); //启动服务器
        if (zk instanceof FollowerZooKeeperServer) {
            FollowerZooKeeperServer fzk = (FollowerZooKeeperServer)zk;
            for(PacketInFlight p: packetsNotCommitted) {
                fzk.logRequest(p.hdr, p.rec);
            }
            for(Long zxid: packetsCommitted) {
                fzk.commit(zxid);
            }
        }
    }

    /**
     * 接收到了leader返回的REVALIDATE信息，进行验证处理
     * @param qp
     * @throws IOException
     */
    protected void revalidate(QuorumPacket qp) throws IOException {
        ByteArrayInputStream bis = new ByteArrayInputStream(qp.getData());
        DataInputStream dis = new DataInputStream(bis);
        long sessionId = dis.readLong();
        boolean valid = dis.readBoolean();
        // 验证完成，从map中移除
        ServerCnxn cnxn = pendingRevalidations.remove(sessionId);
        if (cnxn == null) {
            LOG.warn("Missing session 0x" + Long.toHexString(sessionId) + " for validation");
        } else {
            // 完成session的初始化
            zk.finishSessionInit(cnxn, valid);
        }
    }

    /**
     * learner接收leader的ping命令时，返回LearnerSessionTracker的快照
     * @param qp
     * @throws IOException
     */
    protected void ping(QuorumPacket qp) throws IOException {
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        DataOutputStream dos = new DataOutputStream(bos);
        HashMap<Long, Integer> touchTable = zk.getTouchSnapshot();
        for (Entry<Long, Integer> entry : touchTable.entrySet()) {
            dos.writeLong(entry.getKey());
            dos.writeInt(entry.getValue());
        }
        qp.setData(bos.toByteArray());
        writePacket(qp, true);
    }
    
    
    /**
     * follower感知leader挂了，对zk短时间是灾难性的
     * 1. 其他的follower，会把所有客户端的连接全部断开，
     * 2. 对于客户端，会短时间内频繁报错，内部会不停的尝试去连接其他的zk机器，但是，此时是没有办法重新建立连接的。
     * 3. 此时集群就需要重新进行一个leader选举。
     * 4. 各个follower把状态设置为LOOKING，重新寻找或者选举一个leader出来。
     */
    public void shutdown() {
        // 将zookeeper服务器设置为 null
        self.cnxnFactory.setZooKeeperServer(null);
        // 清除所有连接
        self.cnxnFactory.closeAll();
        // 关闭以前的zookeeper
        if (zk != null) {
            zk.shutdown();
        }
    }
}
